package com.heima.kafka.config;

import lombok.Getter;
import lombok.Setter;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.StreamsConfig;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafkaStreams;
import org.springframework.kafka.annotation.KafkaStreamsDefaultConfiguration;
import org.springframework.kafka.config.KafkaStreamsConfiguration;

import java.util.HashMap;

@Setter
@Getter
@Configuration
@EnableKafkaStreams
@ConfigurationProperties(prefix = "spring.kafka")
public class KafkaStreamConfig {
    private static final int MAX_MESSAGE_SIZE = 16 * 1024 *1024;
    private String hosts;
    private String group;

    @Bean(name = KafkaStreamsDefaultConfiguration.DEFAULT_STREAMS_CONFIG_BEAN_NAME)
    public KafkaStreamsConfiguration defaultKafkaStreamsConfig() {
        HashMap<String, Object> map = new HashMap<>();
        map.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,hosts);                               // Kafka 集群地址
        map.put(StreamsConfig.APPLICATION_ID_CONFIG,this.getGroup() + "_stream_aid");        //应用ID: 每个 Kafka Streams 应用必须有一个唯一的 ID
        map.put(StreamsConfig.CLIENT_ID_CONFIG,this.group + "_stream_cid");                  //客户端ID: 有助于在日志和监控中区分不同的客户端实例
        map.put(StreamsConfig.RETRIES_CONFIG,10);
        map.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        map.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return new KafkaStreamsConfiguration(map);
    }
}
